package org.lql.chaos.rocketmq.service.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.TimeUnit;

/**
 * @author: liangqinglong
 * @date: 2025-08-16 21:10
 * @description: 带监控功能的消息发送回调类
 **/
@Slf4j
public class MonitoredSendCallback implements SendCallback {
    
    // 使用ThreadLocal避免静态变量在多线程环境下的问题
    private static final ThreadLocal<MetricsDelegate> metricsDelegate = new ThreadLocal<MetricsDelegate>() {
        @Override
        protected MetricsDelegate initialValue() {
            return new MetricsDelegate();
        }
    };
    
    private static final ConcurrentHashMap<String, AtomicLong> successCounter = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, AtomicLong> failureCounter = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, AtomicLong> totalLatency = new ConcurrentHashMap<>();
    
    private final String businessType;
    private final long startTime;
    private final String topic;
    
    public MonitoredSendCallback(String businessType, String topic) {
        this.businessType = businessType;
        this.topic = topic;
        this.startTime = System.currentTimeMillis();
    }
    
    @Override
    public void onSuccess(SendResult sendResult) {
        long cost = System.currentTimeMillis() - startTime;
        String key = businessType + ":" + topic;
        
        // 记录成功次数
        successCounter.computeIfAbsent(key, k -> new AtomicLong(0)).incrementAndGet();
        // 记录总延迟
        totalLatency.computeIfAbsent(key, k -> new AtomicLong(0)).addAndGet(cost);
        
        log.info("{}消息发送成功，Topic: {}，MsgId: {}，耗时: {}ms", 
                businessType, topic, sendResult.getMsgId(), cost);
        
        // 上报监控指标
        reportMetrics(key, true, cost);
    }
    
    @Override
    public void onException(Throwable throwable) {
        long cost = System.currentTimeMillis() - startTime;
        String key = businessType + ":" + topic;
        
        // 记录失败次数
        failureCounter.computeIfAbsent(key, k -> new AtomicLong(0)).incrementAndGet();
        
        log.error("{}消息发送失败，Topic: {}，耗时: {}ms，异常信息: {}", 
                businessType, topic, cost, throwable.getMessage(), throwable);
        
        // 上报监控指标
        reportMetrics(key, false, cost);
    }
    
    /**
     * 上报监控指标
     * 在实际项目中，这里可以集成具体的监控系统，如Micrometer、Prometheus等
     * 
     * @param key 监控key
     * @param success 是否成功
     * @param latency 延迟时间
     */
    private void reportMetrics(String key, boolean success, long latency) {
        // 使用Micrometer上报监控指标
        try {
            MetricsDelegate delegate = metricsDelegate.get();
            if (delegate.isMicrometerAvailable()) {
                if (success) {
                    delegate.incrementCounter("rocketmq.send.success", "topic", topic, "business_type", businessType);
                } else {
                    delegate.incrementCounter("rocketmq.send.failure", "topic", topic, "business_type", businessType);
                }
                delegate.recordTimer("rocketmq.send.latency", latency, TimeUnit.MILLISECONDS, "topic", topic, "business_type", businessType);
            }
        } catch (Exception e) {
            log.warn("上报监控指标失败: {}", e.getMessage());
        }
        
        log.debug("上报{}监控指标: key={}, latency={}ms", success ? "成功" : "失败", key, latency);
    }
    
    /**
     * 获取统计信息
     * 
     * @param businessType 业务类型
     * @param topic 主题
     * @return 统计信息字符串
     */
    public static String getStatistics(String businessType, String topic) {
        String key = businessType + ":" + topic;
        AtomicLong success = successCounter.get(key);
        AtomicLong failure = failureCounter.get(key);
        AtomicLong latency = totalLatency.get(key);
        
        long successCount = success != null ? success.get() : 0;
        long failureCount = failure != null ? failure.get() : 0;
        long totalLatencyValue = latency != null ? latency.get() : 0;
        long totalCount = successCount + failureCount;
        double avgLatency = totalCount > 0 ? (double) totalLatencyValue / totalCount : 0;
        
        return String.format("业务类型: %s, 主题: %s, 成功: %d, 失败: %d, 总数: %d, 平均耗时: %.2fms",
                businessType, topic, successCount, failureCount, totalCount, avgLatency);
    }
    
    /**
     * 重置统计信息
     * 
     * @param businessType 业务类型
     * @param topic 主题
     */
    public static void resetStatistics(String businessType, String topic) {
        String key = businessType + ":" + topic;
        AtomicLong success = successCounter.get(key);
        AtomicLong failure = failureCounter.get(key);
        AtomicLong latency = totalLatency.get(key);
        
        if (success != null) {
            success.set(0);
        }
        if (failure != null) {
            failure.set(0);
        }
        if (latency != null) {
            latency.set(0);
        }
    }
    
    /**
     * 内部类用于处理监控指标，避免直接依赖Micrometer类导致的ClassNotFoundException
     */
    private static class MetricsDelegate {
        private Boolean micrometerAvailable = null;
        private Object meterRegistry = null;
        
        public boolean isMicrometerAvailable() {
            if (micrometerAvailable == null) {
                try {
                    Class.forName("io.micrometer.core.instrument.MeterRegistry");
                    Class.forName("io.micrometer.core.instrument.Counter");
                    Class.forName("io.micrometer.core.instrument.Timer");
                    micrometerAvailable = true;
                } catch (ClassNotFoundException e) {
                    micrometerAvailable = false;
                }
            }
            return micrometerAvailable;
        }
        
        public void incrementCounter(String name, String... tags) {
            if (!isMicrometerAvailable()) return;
            
            try {
                if (meterRegistry == null) {
                    // 通过Metrics类获取globalRegistry静态字段
                    Class<?> metricsClass = Class.forName("io.micrometer.core.instrument.Metrics");
                    meterRegistry = metricsClass.getDeclaredField("globalRegistry").get(null);
                }
                
                Class<?> counterClass = Class.forName("io.micrometer.core.instrument.Counter");
                Class<?> registryClass = Class.forName("io.micrometer.core.instrument.MeterRegistry");
                
                Object counter = registryClass.getMethod("counter", String.class, String[].class)
                        .invoke(meterRegistry, name, tags);
                counterClass.getMethod("increment").invoke(counter);
            } catch (Exception e) {
                log.warn("无法增加计数器: {}，详细错误: {}", name, e.getMessage(), e);
            }
        }
        
        public void recordTimer(String name, long amount, TimeUnit unit, String... tags) {
            if (!isMicrometerAvailable()) return;
            
            try {
                if (meterRegistry == null) {
                    // 通过Metrics类获取globalRegistry静态字段
                    Class<?> metricsClass = Class.forName("io.micrometer.core.instrument.Metrics");
                    meterRegistry = metricsClass.getDeclaredField("globalRegistry").get(null);
                }
                
                Class<?> timerClass = Class.forName("io.micrometer.core.instrument.Timer");
                Class<?> registryClass = Class.forName("io.micrometer.core.instrument.MeterRegistry");
                
                Object timer = registryClass.getMethod("timer", String.class, String[].class)
                        .invoke(meterRegistry, name, tags);
                timerClass.getMethod("record", long.class, TimeUnit.class)
                        .invoke(timer, amount, unit);
            } catch (Exception e) {
                log.warn("无法记录计时器: {}，详细错误: {}", name, e.getMessage(), e);
            }
        }
    }
}